package com.corn.kafkastream.demo;

import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.*;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2022/9/15 16:08
 */

public class MergeAndSplitApiTest {
    private static final String BOOTSTRAP_SERVER = "node1:9092";

    private static final String INPUT_TOPIC = "stream.input";

    private static final String OUTPUT_TOPIC = "stream.output";

    private static final String APP_ID = "MERGE_SPLIT_APP_ID";

    public static void main(String[] args) throws InterruptedException {
        // 1. 创建配置
        Properties properties = new Properties();
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID);
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        // 2. streams builder
        StreamsBuilder builder = new StreamsBuilder();

        // split api
        Map<String, KStream<String, String>> branchMap = builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), Serdes.String()).withOffsetResetPolicy(Topology.AutoOffsetReset.LATEST).withName("input-stream"))
                .split(Named.as("splited-ks-"))
                .branch((k, v) -> v.contains("kafka"), Branched.as("kafka-branch"))
                .branch((k, v) -> v.contains("java"), Branched.withFunction(
                        (ks -> ks.mapValues((ValueMapper<String, String>) String::toUpperCase)), "java-branch"))
                .branch((k, v) -> v.contains("haha"), Branched.as("haha-branch"))
                .defaultBranch(Branched.as("default"));

        Thread.sleep(1000);
        // merge
        branchMap.get("splited-ks-kafka-branch")
                .merge(branchMap.get("splited-ks-java-branch"), Named.as("merged-ks"))
                .print(Printed.<String, String>toSysOut().withLabel("merged-data"));

        KafkaStreams streams = new KafkaStreams(builder.build(), properties);

        CountDownLatch countDownLatch = new CountDownLatch(1);

        streams.start();

        countDownLatch.await();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            streams.close();
            countDownLatch.countDown();
        }));

    }
}
